package udf.matching;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.Iterator;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

import udf.string.CleanTag;

public class SplitWithInTags extends EvalFunc<Tuple> {

	private boolean initialized = false;

	private BagFactory bagFactory = BagFactory.getInstance();

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static HashSet<String> seeds = new HashSet<String>();
	//private Hashtable<String, String[]> seeds = new Hashtable<String, String[]>();
	private Hashtable<String,Integer> dictionary = new Hashtable<String,Integer>();
	
	private Hashtable<String,ArrayList<String>> h_map  = new Hashtable<String,ArrayList<String>>(); 
	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";

/**
 * Tuple has the following fields
 * 
 * 0  -  tuple with fields to be intersected
 * 1  -  dictionary to do the look up
 * 2  -  position of tag in dictionary of (1)
 * 3  -  resorting of tuples in , 3 spicify the position in 1
 * 
 * 
 */

		
	// assuming hyphen nonhyphen class
	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(1);
		
		Iterator<Tuple> iterator = bag.iterator();

		while (iterator.hasNext()) {

			Tuple tup = iterator.next();
			
			String h= (String)tup.get(0);
			
			String nh= h.replace("_","");
			
			Integer clase=  Integer.valueOf((String) tup.get(1));
			
			if(h_map.containsKey(nh)){
				
				h_map.get(nh).add(h);
				
			}else{
				
				ArrayList<String> lista = new ArrayList<String>();
				lista.add(h);
				h_map.put(nh, lista);
				
			}
			
			dictionary.put(h,  clase);

		}

		initialized = true;
	}

	public Tuple exec(Tuple input) throws IOException {
		if (!initialized)
			init(input);

		

		Tuple output = TupleFactory.getInstance().newTuple(1);

		String tag = (String) input.get(0);

		ArrayList<String> splitted = null;

		if (h_map.containsKey(tag)) {
			splitted = h_map.get(tag);

		}

		for (int i = 0; splitted!=null && i < splitted.size(); i++) {

			String candidate = splitted.get(i);

			if (dictionary.containsKey(candidate)
					&& dictionary.get(candidate) == 1) {
								
				output.set(0, candidate);
				return output;

			}

		}

		// System.out.println("Processing:" + domain + "\t" + level);
		output.set(0, tag);
		return output;

	}
			
		

}
